// Copyright (c) Microsoft Corporation. All rights reserved.

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;

namespace Microsoft.TeamFoundation.Migration.Toolkit.Wit
{
    /// <summary>
    /// Processing queue class. An abstract queue class that triggers processing once the number
    /// of pending items reaches certain limit.
    /// </summary>
    /// <typeparam name="T">Type of queued items</typeparam>
    class ProcessingQueue<T>
    {
        private QueueConfiguration m_cfg;                   // Configuration
        private ProcessItems m_callback;                    // Callback function
        private List<T> m_items;                            // Queued items

        private object m_lock;                              // Lock object for items
        private Semaphore m_threadSem;                      // Thread semaphore
        private ManualResetEvent m_doneEvent;               // Done event
        private int m_threadCount;                          // Number of active threads

        /// <summary>
        /// Callback function for processing items.
        /// </summary>
        /// <param name="items">Items to process</param>
        /// <param name="queue">Queue the items belonged to</param>
        public delegate void ProcessItems(
            List<T> items,
            ProcessingQueue<T> queue);

        /// <summary>
        /// Returns configuration object.
        /// </summary>
        public QueueConfiguration Config { get { return m_cfg; } }

        /// <summary>
        /// Constructor.
        /// </summary>
        /// <param name="cfg">Configuration</param>
        /// <param name="callback">Callback function</param>
        public ProcessingQueue(
            QueueConfiguration cfg,
            ProcessItems callback)
        {
            m_cfg = cfg;
            m_callback = callback;
            m_items = new List<T>(m_cfg.BatchSize);

            m_lock = new object();
            m_doneEvent = new ManualResetEvent(true);

            if (m_cfg.ThreadCount != 0)
            {
                m_threadSem = new Semaphore(m_cfg.ThreadCount, m_cfg.ThreadCount);
            }
        }

        /// <summary>
        /// Adds an item to the queue.
        /// </summary>
        /// <param name="item">Item to add</param>
        public void Add(
            T item)
        {
            lock (m_lock)
            {
                m_items.Add(item);
                if (m_items.Count == m_cfg.BatchSize)
                {
                    Flush();
                }
            }
        }

        /// <summary>
        /// Adds pending items, if any, to the given list.
        /// </summary>
        /// <param name="items">List of items</param>
        public void AppendPendingItems(
            List<T> items)
        {
            if (items.Count < m_cfg.BatchSize)
            {
                lock (m_lock)
                {
                    int count = Math.Min(m_cfg.BatchSize - items.Count, m_items.Count);

                    if (count > 0)
                    {
                        items.AddRange(m_items.GetRange(0, count));
                        m_items.RemoveRange(0, count);
                    }
                }
            }
        }

        /// <summary>
        /// Flushes content of the queue.
        /// </summary>
        /// <param name="waitFlag">Tells to wait until all items have been processed</param>
        public void Flush()
        {
            List<T> items;

            lock (m_lock)
            {
                if (m_items.Count > 0)
                {
                    items = m_items;
                    m_items = new List<T>(m_cfg.BatchSize);
                }
                else
                {
                    items = null;
                }
            }

            if (items != null)
            {
                if (m_cfg.ThreadCount == 0)
                {
                    m_callback(items, this);
                    m_doneEvent.Set();
                }
                else
                {
                    m_threadSem.WaitOne();
                    try
                    {
                        Interlocked.Increment(ref m_threadCount);
                        m_doneEvent.Reset();
                        Thread t = new Thread(new ParameterizedThreadStart(ThreadProc));
                        t.Name = "Processing thread";
                        t.Start(items);
                    }
                    finally
                    {
                        m_threadSem.Release();
                    }
                }
            }
        }

        /// <summary>
        /// Flushes content and waits until processing stops.
        /// </summary>
        internal void FlushAndWait()
        {
            Flush();
            m_doneEvent.WaitOne();
        }

        /// <summary>
        /// Returns "done" event.
        /// </summary>
        internal WaitHandle DoneEvent { get { return m_doneEvent; } }

        /// <summary>
        /// Thread procedure.
        /// </summary>
        /// <param name="list">List of items</param>
        private void ThreadProc(
            object list)
        {
            m_threadSem.WaitOne();
            try
            {
                // Process items
                List<T> items = (List<T>)list;
                Debug.Assert(items.Count > 0, "No items to process!");
                Debug.Assert(items.Count <= m_cfg.BatchSize, "Too many items to process!");

                m_callback(items, this);
            }
            finally
            {
                m_threadSem.Release();
            }

            if (Interlocked.Decrement(ref m_threadCount) == 0)
            {
                m_doneEvent.Set();
            }
        }
    }
}
